package org.obp.ons.client;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import org.obp.ons.config.OnsConfig;
import org.obp.ons.consts.MqMessageType;
import org.obp.ons.group.MqGroups;
import org.obp.ons.router.MqMessageRouter;
import org.obp.ons.topic.MqTopics;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Configuration
public class MqConsumerClient implements InitializingBean {

    @Autowired
    private MqTopics mqTopics;
    @Autowired
    private MqGroups mqGroups;
    @Autowired
    private OnsConfig onsConfig;

    @Autowired
    private MqMessageRouter mqMessageRouter;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnProperty(name = "ons.rocketmq.enable", havingValue = "true")
    public ConsumerBean buildConsumer() {

        ConsumerBean consumerBean = new ConsumerBean();

        // 配置文件
        Properties properties = onsConfig.getMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, mqGroups.getGroupId());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(properties);

        // 订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        Subscription subscription = new Subscription();
        subscription.setTopic(mqTopics.getTopic());
        subscription.setExpression(MqMessageType.NORMAL.name());
        subscriptionTable.put(subscription, mqMessageRouter);
        consumerBean.setSubscriptionTable(subscriptionTable);

        return consumerBean;
    }

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    @ConditionalOnProperty(name = "ons.rocketmq.enable", havingValue = "true")
    public ConsumerBean buildTimingConsumer() {

        ConsumerBean consumerBean = new ConsumerBean();

        // 配置文件
        Properties properties = onsConfig.getMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, mqGroups.getGroupId());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(properties);

        // 订阅关系
        Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>();
        Subscription subscription = new Subscription();
        subscription.setTopic(mqTopics.getTopic_timing());
        subscription.setExpression(MqMessageType.TIMING.name());
        subscriptionTable.put(subscription, mqMessageRouter);
        consumerBean.setSubscriptionTable(subscriptionTable);

        return consumerBean;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

    }

}
